package com.roncoo.eshop.storm.bolt;

import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;

/**
 * Copyright: Copyright (c) 2019 Hu-Cheng
 *
 * @ClassName: LogParseBolt.java
 * @Description: 接收spout数据，解析日志的bolt
 *
 * @version: v1.0.0
 * @author: hucheng
 * @date: Oct 15, 2019 1:41:49 PM
 *
 */
@SuppressWarnings("serial")
public class LogParseBolt extends BaseRichBolt {

	@SuppressWarnings("unused")
	private static final long SerialVersionUID = -8761807561458126413L;

	private OutputCollector collector;

	private static final Logger LOGGER = LoggerFactory.getLogger(LogParseBolt.class);

	@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
	}

	@Override
	public void execute(Tuple input) {
		String message = input.getStringByField("message");
		JSONObject messageJsonObject = JSONObject.parseObject(message);
		JSONObject uriArgsJsonObject = messageJsonObject.getJSONObject("uri_args");

		Long productId = uriArgsJsonObject.getLong("productId");

		LOGGER.info("接收spout数据，解析日志的bolt，productId = " + productId);

		if (productId != null) {
			collector.emit(new Values(productId));
		}
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("productId"));
	}

}
